/*
 * Copyright (c) 2020-2099 the original author or authors. All rights reserve.
 */
package org.tianyun.cloud.amqp.rabbitmq.message;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback;
import org.tianyun.cloud.context.RequestContext.RequestContextHolder;
import org.tianyun.cloud.context.ServletContext.ServletContextHolder;

import java.util.Map;

/**
 * RabbitMQ消息发送方消息确认，处理ConfirmCallback、ReturnsCallback
 *
 * @auther ebert_chan
 */
public class RabbitMessageConfirmAndReturn implements ConfirmCallback, ReturnsCallback {

    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMessageConfirmAndReturn.class);

    public void confirmCallback(CorrelationData correlationData, String cause) {
        LOGGER.warn("消息投递到Exchange失败: {}", cause);
    }

    public void returnsCallback(ReturnedMessage returned) {
        LOGGER.warn("消息投递到Queue失败: {}", returned);
    }

    /*
     * @see org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback#confirm(org.springframework.amqp.rabbit.connection.CorrelationData,
     * boolean, java.lang.String)
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            return;
        }
        Message message = correlationData != null && correlationData.getReturned() != null ? correlationData.getReturned().getMessage() : null;
        if (message != null) {
            this.setContext(message);
        }
        try {
            this.confirmCallback(correlationData, cause);
        } finally {
            ServletContextHolder.clear();
        }
    }

    /*
     * @see org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback#returnedMessage(org.springframework.amqp.core.ReturnedMessage)
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        Message message = returned.getMessage();
        if (message != null) {
            this.setContext(message);
        }
        try {
            this.returnsCallback(returned);
        } finally {
            ServletContextHolder.clear();
        }
    }

    private void setContext(Message message) {
        MessageProperties messageProperties = message.getMessageProperties();
        if (messageProperties == null) {
            return;
        }
        Map<String, Object> headers = messageProperties.getHeaders();
        RequestContextHolder.setContextMap(headers);
        ServletContextHolder.setContextMap(headers);
    }

}